package com.study.service;

import com.study.entity.Order;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

@Service
public class HbaseOrderService {
    @Autowired
    private Connection hbaseConnection;

    //插入数据
    public void insert(Order order){
        try {
            //获取数据库中的表
            Table table = null;
            table = hbaseConnection.getTable(TableName.valueOf("order"));
            //准备一行数据
            Put line = new Put(order.getId().getBytes());
            line.addColumn("orderInfo".getBytes(),"orderNo".getBytes(),order.getOrderNo().getBytes());
            line.addColumn("orderInfo".getBytes(),"orderType".getBytes(),order.getOrderType().getBytes());
            line.addColumn("orderInfo".getBytes(),"orderTime".getBytes(),order.getOrderTime().getBytes());

            line.addColumn("expressInfo".getBytes(),"expressCom".getBytes(),order.getExpressCom().getBytes());

            //将数据插入数据库
            table.put(line);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    //插入数据
    public String delete(String rowkey){
        try {
            Table table = hbaseConnection.getTable(TableName.valueOf("order"));
            Delete line = new Delete(rowkey.getBytes());
            table.delete(line);
            return "sccess";
        } catch (IOException e) {
            e.printStackTrace();
        }
        return "false";
    }

    //根据rowkey获取数据
    public Order findByRowkey(String rowkey){
        Table table = null;
        Result result = null;
        Order us = null;
        //获取hbase中的表
        try {
            table = hbaseConnection.getTable(TableName.valueOf("order"));
            //按照rowkey获取数据
            Get get = new Get(rowkey.getBytes());
            result = table.get(get);
            us = Order.builder().orderNo(new String(result.getValue("orderInfo".getBytes(),"orderNo".getBytes()))).build();
            us.setId(rowkey);
            us.setOrderTime(new String(result.getValue("orderInfo".getBytes(),"orderTime".getBytes())));
            us.setOrderType(new String(result.getValue("orderInfo".getBytes(),"orderType".getBytes())));
            us.setExpressCom(new String(result.getValue("expressInfo".getBytes(),"expressCom".getBytes())));
        } catch (IOException e) {
            e.printStackTrace();
        }finally {
            try {
                table.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return us;
    }


    //代码逻辑没问题  在海量数据下此方法不能使用 内存不足 会花费大量时间下载数据
    public List<Order> findAll(){
        List<Order> list = new ArrayList<Order>();
        Table table = null;
        try {
            table = hbaseConnection.getTable(TableName.valueOf("order"));
            Scan scan = new Scan();
            ResultScanner rs = table.getScanner(scan);
            Result result = null;
            while ((result =rs.next())!= null){
                list.add(Order.builder().orderNo(new String(result.getValue("orderInfo".getBytes(),"orderNo".getBytes()))).build());
            }
        } catch (IOException e) {
            e.printStackTrace();
        }finally {
            try {
                table.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return list;
    }

    //根据username包含某字查询信息
    public List<Order> findUserByPrefixName(String prefixName){
        Table table = null;
        List<Order> list = new ArrayList<>();
        try {
            table = hbaseConnection.getTable(TableName.valueOf("order"));
            Scan scan = new Scan();
            //hbase中操纵命令：scan 'Order',{FILTER=>"SingleColumnValueFilter('base','username',=,'substring:zhang')"}
            SingleColumnValueFilter vf = new SingleColumnValueFilter(
                    "orderInfo".getBytes(),
                    "username".getBytes(),
                    CompareFilter.CompareOp.EQUAL,
                    new SubstringComparator(prefixName)
            );
            scan.setFilter(vf);
            ResultScanner scanner = table.getScanner(scan);

            Result rs = null;
            while ((rs=scanner.next()) != null){
                Order order = Order.builder().orderNo(new String(rs.getValue("orderInfo".getBytes(), "orderNo".getBytes()))).build();
                order.setId(new String(rs.getRow()));
                order.setOrderTime(new String(rs.getValue("orderInfo".getBytes(),"orderTime".getBytes())));
                order.setOrderType(new String(rs.getValue("orderInfo".getBytes(),"orderType".getBytes())));
                order.setExpressCom(new String(rs.getValue("expressInfo".getBytes(),"expressCom".getBytes())));
                list.add(order);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                table.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return list;
    }
}
